package com.example.mymessagequeue.mqserver.datacenter;

import com.example.mymessagequeue.common.BinaryTool;
import com.example.mymessagequeue.common.MqException;
import com.example.mymessagequeue.mqserver.core.Message;
import com.example.mymessagequeue.mqserver.core.MsgQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/**
 * 此类是对硬盘上的消息进行管理
 */
public class MessageFileManager {

    // 定义一个内部类, 表示该队列的统计信息
    static public class Stat {
        public int totalCount; // 总消息数量
        public int validCount; // 有效消息数量
    }

    public void init() {
        // 暂时不需要做额外的初始化工作, 方便后续扩展
    }

    // 约定消息文件所在的目录和文件名

    /**
     * 获取指定队列对应的消息所在路径
     *
     * @param  queueName
     * @return
     */
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    /**
     * 获取该队列的消息数据文件路径
     *
     * @param queueName
     * @return
     */
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    /**
     * 获取队列的消息统计文件路径
     *
     * @param queueName
     * @return
     */
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    private Stat readStat(String queueName) {
        // 当前消息统计文件是文本文件, 可以直接使用 Scanner 读取文件
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    private void writeStat(String queueName, Stat stat) {
        // 使用 PrintWrite 来写文件
        // OutputStream 打开文件, 默认会直接把原文件清空
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建队列对应的文件和目录
     *
     * @param queueName
     */
    public void createQueueFiles(String queueName) throws IOException {
        // 1. 先创建队列对应的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("创建目录失败! baseDir=" + baseDir.getAbsolutePath());
            }
        }
        // 2. 创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }
        // 3. 创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }
        // 4. 给消息统计文件设计初始值: 0\t0
        Stat stat = new Stat();
        stat.validCount = 0;
        stat.totalCount = 0;
        writeStat(queueName, stat);
    }

    /**
     * 删除队列的目录和文件, 队列也是可以被创建和删除的
     *
     * @param queueName
     */
    public void destroyQueueFiles(String queueName) throws IOException {
        // 1. 删除里面的文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStateFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStateFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if (!ok1 || !ok2 || !ok3) {
            // 任何一个删除失败, 都算删除失败
            throw new IOException("删除队列目录和文件失败！ baseDir=" + baseDir.getAbsolutePath());
        }
    }

    /**
     * 队列的目录和文件是否存在(数据文件和统计文件)
     *
     * @param queueName
     * @return
     */
    public boolean checkFilesExits(String queueName) {
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            return false;
        }
        File queueStat = new File(getQueueStatPath(queueName));
        if (!queueStat.exists()) {
            return false;
        }
        return true;
    }

    /**
     * 把一个新的消息，放到队列对应的文件中
     *
     * @param queue
     * @param message
     */
    public void sendMessage(MsgQueue queue, Message message) throws MqException, IOException {
        // 1. 检查当前要写入的队列对应的文件是否存在
        if (!checkFilesExits(queue.getName())) {
            throw new MqException("[MessageFileManager] 队列对应的文件不存在! queueName=" + queue.getName());
        }

        // 2. 把 Message 对象进行序列化, 转成二进制数组
        byte[] messageBinary = BinaryTool.toBytes(message);

        synchronized (queue) {
            // 3. 获取当前队列数据文件的长度, 用这个来计算出该 Message 对象的 offsetBeg 和 offsetEnd
            // 把新的 Message, 写入到队列数据文件的末尾, 此时 Message 对象的 offsetBeg, 就是当前文件长度 + 4
            // offsetEnd 就是当前文件长度 + 4 + message 自身长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            // 文件长度 queueDataFile.length();
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);

            // 4. 写入消息到文件(追加)
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                // 先写入消息的长度, 占据四个字节
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    dataOutputStream.writeInt(messageBinary.length);
                    // 写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }

            // 5. 更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }
    }

    /**
     * 删除消息 - 逻辑删除
     *
     * @param queue
     * @param message
     */
    public void deleteMessage(MsgQueue queue, Message message) throws IOException, ClassNotFoundException {
        synchronized (queue) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                // 1. 从文件中读取对应的 Message 数据
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg()); // 光标指到 beg 位置
                randomAccessFile.read(bufferSrc); // 把 bufferSrc 的空间读满

                // 2. 把当前读出来的二进制数据转换成 Message 对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);

                // 3. 把 isValid 设置无效
                diskMessage.setIsValid((byte) 0x0);

                // 4. 重新写入文件
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                randomAccessFile.seek(message.getOffsetBeg()); // 因为 read 和 write 都会移动光标, 所以这里需要重新移动光标到起始位置
                randomAccessFile.write(bufferDest);
            }

            // 5. 更新消息统计文件
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0) {
                stat.validCount -= 1;
            }
            writeStat(queue.getName(), stat);
        }
    }


    /**
     * 从文件中读取出所有的消息内容, 这个方法, 准备在程序启动的时候进行调用
     * 只返回有效的消息
     * @param queueName 这里不传 MsgQueue 对象, 因为是读操作, 不需要加锁, 防止锁冲突 + 因为这个方法是在程序启动时调用, 此时咱们服务器还不能处理请求, 所以不涉及多线程操作文件
     * @return
     */
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                long currentOffset = 0;  // 记录当前文件光标
                // 一个文件中包含很多消息, 所以进行循环读
                while (true) {
                    // 1. 读取Message长度, 这里的 readInt 可能会读到文件的末尾, 看这个方法注释
                    // readInt() 方法读到文件末尾, 会抛出 EOFException 异常, 这点和其他很多流对象不同
                    int messageSize = dataInputStream.readInt();

                    // 2. 按照消息长度读取消息的内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if (messageSize != actualSize) {
                        // 如果不匹配, 说明文件有问题, 格式错乱了
                        throw new MqException("[MessageFileManager] 文件格式错误! queueName=" + queueName);
                    }

                    // 3. 把读到的二进制数据, 反序列化成 Message 对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);

                    // 4.判定这个消息对象是否是无效对象
                    if (message.getIsValid() != 0x1) {
                        currentOffset += (4 + messageSize);  // 光标后移
                        continue;
                    }

                    // 5. 有效数据, 则先填写 offsetBeg 和 offsetEnd
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);  // 光标后移

                    // 6. 将 Message 对象加入到链表中
                    messages.add(message);
                }
            } catch (EOFException e) {
                // 这个 catch 不是真正处理异常, 而是正常处理业务逻辑, 文件读到末尾, 会被 readInt 抛出该异常.
                System.out.println("[MessageFileManager] 恢复 Message 数据完成!");
            }
        }
        return messages;
    }


    /**
     * 检测当前是否需要垃圾回收
     *
     * @param queueName
     * @return
     */
    public boolean checkGC(String queueName) {
        // 判定是否要GC: 总消息数 > 2000, 有效消息数小于总消息的一半
        Stat stat = readStat(queueName);
        return (stat.totalCount > 2000 && (double) stat.validCount / (double) stat.totalCount < 0.5);
    }

    /**
     * 获取新文件的路径
     *
     * @param queueName
     * @return
     */
    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    /**
     * 实现指定队列的消息数据文件的垃圾回收
     * 使用 复制算法 来完成
     * 主要逻辑：创建一个新的文件, 名字是 queue_data_new.txt
     * 把之前消息数据文件中的有效数据读取出来, 写入到新的文件中
     * 删除旧的文件, 把新文件改名回 queue_data.txt
     * 更新消息统计文件
     *
     * @param queue
     */
    public void gc(MsgQueue queue) throws MqException, IOException, ClassNotFoundException {
        // 进行 gc 时, 会涉及到对消息文件进行删除和修改, 所以此时其他线程不能针对当前消息文件做任何事情
        synchronized (queue) {
            // 由于 gc 操作可能比较耗时, 此时统计下执行消耗时间
            long gcBeg = System.currentTimeMillis();

            // 1. 创建一个新的文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNewFile.exists()) {
                // 正常情况下这个文件不应该存在, 如果存在, 说明上次 gc 一半, 程序意外崩溃了
                throw new MqException("[MessageFileManager] gc 时意外发现该队列的 queue_data_new.txt 文件已经存在！ " +
                        "queueName=" + queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if (!ok) {
                throw new MqException("[MessageFileManager] 创建 queue_data_new.txt 文件失败！ " +
                        "queueDataNewFile=" + queueDataNewFile.getAbsolutePath());
            }

            // 2. 从旧文件中, 读取出所有的有效消息对象
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());

            // 3. 把有效消息, 写入到新的文件中.
            // 此时不循环调用 sendMessage() 是因为此时是往 queue_data_new.txt 文件中写
            try (OutputStream outputStream = new FileOutputStream(queueDataNewFile)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages) {
                        byte[] buffer = BinaryTool.toBytes(message);
                        // 内存上的 Message对应的 offsetBeg 和 offsetEnd 也得修改，不然下次就找不到这个硬盘上的消息了
                        // 这步操作交给 VirtualHost 去做, 只有它能同时操作硬盘和内存的数据
                        dataOutputStream.writeInt(buffer.length); // 先写四个字节表示消息长度
                        dataOutputStream.write(buffer);  // 写入消息内容
                    }
                }
            }

            // 4. 删除旧的数据文件, 并且把新的文件进行重命名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if (!ok) {
                throw new MqException("[MessageFileManager] 删除旧的 queue_data.txt 文件失败！ " +
                        "queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
            }
            // 命名新文件 queue_data_new.txt -> queue_data.txt
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if (!ok) {
                throw new MqException("[MessageFileManager] 文件重命名失败！ queueDataNewFile=" +
                        queueDataNewFile.getAbsolutePath() + "queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
            }

            // 5. 更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(), stat);

            long gcEed = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc 执行完毕! queueName=" + queue.getName()
                    + ", time=" + (gcEed - gcBeg) + "ms");
        }
    }
}
